﻿using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Confluent.Kafka;
using Confluent.Kafka.Admin;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using NetMicro.Core.Extensions;
using NetMicro.EventBus.Enums;
using NetMicro.EventBus.Extensions;
using NetMicro.EventBus.Messages;

namespace NetMicro.EventBus.Kafka
{
    public class KafkaEventSubscriber : IEventSubscriber
    {
        public KafkaEventSubscriber(IKafkaConnection connection,
            IMessageSerializer messageSerializer,
            ILogger<KafkaEventSubscriber> logger,
            IMessageListener messageListener,
            IEventErrorHandler eventErrorHandler,
            IOptions<EventBusKafkaOptions> options)
        {
            Connection = connection;
            MessageSerializer = messageSerializer;
            Logger = logger;
            MessageListener = messageListener;
            EventErrorHandler = eventErrorHandler;
            Options = options.Value;
        }
        private EventBusKafkaOptions Options { get; }
        private IKafkaConnection Connection { get; }
        private IMessageSerializer MessageSerializer { get; }
        private ILogger<KafkaEventSubscriber> Logger { get; }
        private IMessageListener MessageListener { get; }

        private IEventErrorHandler EventErrorHandler { get; }

        public async Task SubscribeAsync(EventHandlerDescriptor eventHandlerDescriptor, CancellationToken cancellationToken)
        {

            if (cancellationToken.IsCancellationRequested)
                return;
            string eventName = eventHandlerDescriptor.EventName;
            string eventHandlerName = eventHandlerDescriptor.EventHandlerName;
            IConsumer<string, byte[]> consumer = Connection.CreateConsumer(eventHandlerName);
            consumer.Subscribe(eventName);

            _ = Task.Run(async () =>
            {
                while (!cancellationToken.IsCancellationRequested)
                {
                    try
                    {
                        await Consume(consumer, eventName, eventHandlerName, cancellationToken);
                    }
                    catch (AccessViolationException)
                    {
                        // ignore
                    }
                    catch (Exception e)
                    {
                        Logger.LogError(e, $"[EventBus] consume message occur error.");
                    }

                    // ReSharper disable once MethodSupportsCancellation
                    await Task.Delay(5).ConfigureAwait(false);
                }
            }, cancellationToken);
        }

        private async Task Consume(IConsumer<string, byte[]> consumer, string eventName, string eventHandlerName, CancellationToken cancellationToken)
        {
            if (cancellationToken.IsCancellationRequested)
            {
                return;
            }

            ConsumeResult<string, byte[]> consumerResult;
            try
            {
                consumerResult = consumer.Consume(cancellationToken);
                if (consumerResult.IsPartitionEOF || consumerResult.Message.Value.IsNullOrEmpty())
                    return;
            }
            catch (OperationCanceledException)
            {
                return;
            }
            catch (Exception ex)
            {
                Logger.LogError(ex,
                    $"[EventBus-Kafka] consume message occur error, event: {eventName}, handler: {eventHandlerName}.");
                return;
            }

            MessageTransfer message;
            try
            {
                message = MessageSerializer.DeserializeE<MessageTransfer>(consumerResult.Message.Value);
            }
            catch (Exception e)
            {
                Logger.LogError(e, "[EventBus-Kafka] deserialize message from kafka error.");
                return;
            }

            if (message is null)
            {
                Logger.LogError("[EventBus-Kafka] deserialize message from kafka error.");
                return;
            }

            if (message.EventName != eventName)
            {
                Logger.LogError(
                    $"[EventBus-Kafka] received invalid event name \"{message.EventName}\", expect \"{eventName}\".");
                return;
            }

            Logger.LogDebug(
                $"[EventBus-Kafka] received msg: {message.ToJson()}.");
            message.EventHandlerName = eventHandlerName;
            // 执行消息监听处理
            MessageReceiveResult res = await MessageListener.OnReceiveAsync(message, cancellationToken).ConfigureAwait(false);

            if (res.MessageReceive == MessageReceive.Failed)
            {
                message.Exception = res.ExceptionMsg;
                await EventErrorHandler.OnErrorAsync(message, cancellationToken).ConfigureAwait(false);
            }
            consumer.Commit(consumerResult);
        }
    }
}